package com.ytoglobal.overseas.address.platform.dao;

import com.ytoglobal.overseas.address.platform.common.exception.AppException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeRequest;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
@Slf4j
public class ElasticsearchRestDao {

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    /**
     * 新增，修改文档
     *
     * @param indexName 索引
     * @param type      mapping type
     * @param id        文档id
     * @param jsonStr   文档数据
     */
    public void addData(String indexName, String type, String id, String jsonStr) {
        try {
            // 1、创建索引请求  //索引  // mapping type  //文档id
            IndexRequest request = new IndexRequest(indexName, type, id);     //文档id
            // 2、准备文档数据
            // 直接给JSON串
            request.source(jsonStr, XContentType.JSON);
            //4、发送请求
            IndexResponse indexResponse = null;
            try {
                // 同步方式
                indexResponse = restHighLevelClient.index(request);
            } catch (ElasticsearchException e) {
                log.error("ES写入数据失败！", e);
            }
            //5、处理响应
            if (indexResponse != null) {
                String index1 = indexResponse.getIndex();
                String type1 = indexResponse.getType();
                String id1 = indexResponse.getId();
                long version1 = indexResponse.getVersion();
                if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
                    log.info("新增文档成功!" + index1 + type1 + id1 + version1);
                } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
                    log.info("修改文档成功!");
                }
                // 分片处理信息
                ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
                if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                    log.info("分片处理信息.....");
                }
                // 如果有分片副本失败，可以获得失败原因信息
                if (shardInfo.getFailed() > 0) {
                    for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                        String reason = failure.reason();
                        log.error("副本失败原因：" + reason);
                    }
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 批量插入ES
     *
     * @param indexName 索引
     * @param type      类型
     * @param idName    id名称
     * @param list      数据集合
     */
    public void bulkAddDate(String indexName, String type, String idName, List<Map<String, Object>> list) {
        try {

            if (null == list || list.size() <= 0) {
                return;
            }
            if (StringUtils.isBlank(indexName) || StringUtils.isBlank(type)) {
                return;
            }
            BulkRequest request = new BulkRequest();
            for (Map<String, Object> map : list) {
                if (idName == null) {
                    request.add(new IndexRequest(indexName, type).source(map, XContentType.JSON));
                } else {
                    if (map.get(idName) != null) {
                        request.add(new IndexRequest(indexName, type, String.valueOf(map.get(idName))).source(map, XContentType.JSON));
                    }
                }
            }
            // 同步请求
            BulkResponse bulkResponse = restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
            // 处理响应
            if (bulkResponse != null) {
                for (BulkItemResponse bulkItemResponse : bulkResponse) {
                    DocWriteResponse itemResponse = bulkItemResponse.getResponse();

                    if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX
                            || bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) {
                        IndexResponse indexResponse = (IndexResponse) itemResponse;
                        log.info("新增成功,{}" + indexResponse.toString());
                    } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) {
                        UpdateResponse updateResponse = (UpdateResponse) itemResponse;
                        log.info("修改成功,{}" + updateResponse.toString());
                    } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) {
                        DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
                        log.info("删除成功,{}" + deleteResponse.toString());
                    }
                }
            }
        } catch (IOException e) {
            log.error("批量插入ES数据失败", e);
        }
    }

    /**
     * 查询ES数据
     *
     * @param index               es 索引
     * @param type                es 索引类型（es 7.0版本取消掉type）
     * @param searchSourceBuilder es查询条件
     * @return 查询结果
     */
    public List<Map<String, Object>> searchIndex(String index, String type, SearchSourceBuilder searchSourceBuilder) {
        List<Map<String, Object>> result = new ArrayList<>();
        SearchRequest searchRequest = new SearchRequest(index);
        searchRequest.types(type);
        searchRequest.source(searchSourceBuilder);
        try {
            SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            SearchHits hits = response.getHits();
            SearchHit[] searchHits = hits.getHits();
            for (SearchHit hit : searchHits) {
                result.add(hit.getSourceAsMap());
            }
        } catch (IOException e) {
            log.error("查询ES数据失败", e);
        }
        return result;
    }

    /**
     * 查询ES数据
     *
     * @param index               es 索引
     * @param type                es 索引类型（es 7.0版本取消掉type）
     * @param searchSourceBuilder es查询条件
     * @return 查询结果
     */
    public Map<String, Object> searchIndexForOneResult(String index, String type, SearchSourceBuilder searchSourceBuilder) {
        Map<String, Object> result = new HashMap<>();
        SearchRequest searchRequest = new SearchRequest(index);
        searchRequest.types(type);
        searchRequest.source(searchSourceBuilder);
        try {
            SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            SearchHits hits = response.getHits();
            SearchHit[] searchHits = hits.getHits();
            if (searchHits.length > 0) {
                result.putAll(searchHits[0].getSourceAsMap());
            }
        } catch (IOException e) {
            log.error("查询ES数据失败", e);
        }
        return result;
    }

    /*
     * index 索引index
     * text 需要被分析的词语
     * 默认使用中文ik_smart分词
     * */
    public String[] getAnalyzes(String index, String text) {
        AnalyzeRequest analyzeRequest = new AnalyzeRequest(index);
        analyzeRequest.analyzer("ik_smart");
        analyzeRequest.text(text);
        try {
            AnalyzeResponse analyzeResponse = restHighLevelClient.indices().analyze(analyzeRequest, RequestOptions.DEFAULT);
            List<AnalyzeResponse.AnalyzeToken> analyzeTokenList = analyzeResponse.getTokens();
            // 赋值
            List<String> searchTermList = new ArrayList<>();
            analyzeTokenList.forEach(ikToken -> searchTermList.add(ikToken.getTerm()));
            return searchTermList.toArray(new String[searchTermList.size()]);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return new String[0];
    }

    public Map<String, List<String>> getMatchAggsData(String index, SearchSourceBuilder searchSourceBuilder) {
        Map<String, List<String>> aggDataMap = new HashMap<>();
        SearchRequest searchRequest = new SearchRequest(index);
        searchRequest.source(searchSourceBuilder);
        try {
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            Aggregations aggregations = searchResponse.getAggregations();
            Map<String, Aggregation> aggregationMap = aggregations.getAsMap();
            aggregationMap.forEach((aggKey, key) -> {
                List<String> list = new ArrayList<>(1);
                ParsedStringTerms terms = aggregations.get(aggKey);
                List<? extends Terms.Bucket> buckets = terms.getBuckets();
                for (Terms.Bucket bucket : buckets) {
                    String keyName = bucket.getKey().toString();
                    list.add(keyName);
                }
                aggDataMap.put(aggKey, list);
            });
        } catch (IOException e) {
            log.error(index + "聚合查询失败", e);
            throw new AppException(500, "聚合失败");
        }
        return aggDataMap;
    }

    public void deleteIndex(String index) {
        DeleteIndexRequest request = new DeleteIndexRequest(index);
        request.timeout(TimeValue.timeValueMinutes(2)); //设置超时，等待所有节点确认索引删除（使用TimeValue形式）
        //设置IndicesOptions控制如何解决不可用的索引以及如何扩展通配符表达式
        request.indicesOptions(IndicesOptions.lenientExpandOpen());
        try {
            restHighLevelClient.indices().delete(request,RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error(index + "删除索引失败", e);
        }
    }
}
